package com.ionehe.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.ionehe.annotation.ElasticJobConfig;
import com.ionehe.enums.BrokerMessageStatus;
import com.ionehe.model.BrokerMessage;
import com.ionehe.domain.service.RabbitBroker;
import com.ionehe.domain.service.impl.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@ElasticJobConfig(
        name = "com.ionehe.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?", //10秒钟一次
        description = "可靠性投递消息补偿任务",
        overwrite = true,
        shardingTotalCount = 1  //broker_message表只有一张，加入有100张表，可以采用10个分片，每个分片处理10张表
)
@Slf4j
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {
 
    @Autowired
    private MessageStoreService messageStoreService;
 
    @Autowired
    private RabbitBroker rabbitBroker;
 
    private static final int MAX_RETRY_COUNT = 3;
 
    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> list = messageStoreService.fetchTimeOutMessage4Retry(BrokerMessageStatus.SENDING);
        log.info("------@@@抓取的数据数量：{}", list.size());
        return list;
    }
 
    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> list) {
        list.forEach(brokerMessage -> {
            if(brokerMessage.getTryCount() >= MAX_RETRY_COUNT){
                this.messageStoreService.failure(brokerMessage.getMessageId());
                log.warn("消息重试最终失败，消息设置为最终失败，消息Id={}", brokerMessage.getMessageId());
            }else {
                //每次重发的时候更新try_count字段 ，值+1
                this.messageStoreService.updateTryCount(brokerMessage.getMessageId());
                //重发消息
                this.rabbitBroker.reliantSend(brokerMessage.getMessage());
            }
        });
    }
}